package com.prac.mq.original;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.junit.Test;

/**
 * 含标签的消息使用
 *
 * @author H
 * @version 1.0
 * @date 2023-11-10 10:42
 */
@Slf4j
public class MqTagTest {

    @Test
    public void tagProducer() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("tag-producer", false);
        producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR2);
        //producer.setMqClientApiTimeout(1000 * 10);
        producer.start();
        for (int i = 0; i < 10; i++) {
            Message message = new Message("tag-topic", "order-tag", (i + "====>" + "order信息").getBytes());
            Message message1 = new Message("tag-topic", "trans-tag", (i + "====>" + "trans信息").getBytes());
            Message message2 = new Message("tag-topic", "pay-tag", (i + "====>" + "pay信息").getBytes());

            producer.send(message, 1000 * 10);
            producer.send(message1, 1000 * 10);
            producer.send(message2, 1000 * 10);
        }
        log.info("消息投递成功");
        producer.shutdown();
    }

    @Test
    public void tagConsumer() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("tag-consumer");
        consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR2);
        consumer.subscribe("tag-topic", "pay-tag");
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            log.info("当前线程：" + Thread.currentThread().getId() + "===>" + new String(msgs.get(0).getBody()));
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
        System.in.read();
    }

    @Test
    public void tagConsumer2() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("tagConsumer-b");
        consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR2);
        consumer.subscribe("tag-topic", "order-tag||pay-tag");
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            log.info("当前线程：" + Thread.currentThread().getId() + "===>" + new String(msgs.get(0).getBody()));
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
        System.in.read();
    }

}
